package com.qupeng.demo.kafka.kafkaspringbootconsumer.listener.concurrency;

import com.qupeng.demo.kafka.kafkaspringbootconsumer.model.Product;
import com.qupeng.demo.kafka.kafkaspringbootconsumer.config.KafkaProperties;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
@ConditionalOnProperty(name = "app.kafka.consumer.mode.batch-listener", havingValue = "true", matchIfMissing = false)
public class BatchKafkaListener {

    @Autowired
    private KafkaProperties kafkaProperties;

    @KafkaListener(id = "batch-listener",
            groupId = "#{kafkaProperties.getGroupId()}",
            topics = "#{kafkaProperties.getTopics()}",
            clientIdPrefix = "batch-listener",
            containerFactory = "batchFactory",
            properties = {}
    )
    public void batchListen(ConsumerRecords<Integer, Product> consumerRecords) {
        consumerRecords.forEach(System.out::println);
    }

}

